package org.jackkang.elasticsearch.util;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

import javax.annotation.Resource;

import org.jackkang.elasticsearch.model.News;
import org.jackkang.elasticsearch.service.NewsService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.data.elasticsearch.core.query.IndexQuery;
import org.springframework.data.elasticsearch.core.query.IndexQueryBuilder;

/**
 * 
 * @ClassName: NewsSyncService
 * @Description: 同步新闻信息
 * @author  kangjie1209@163.com
 * @date 2015年10月20日 上午11:20:16
 *
 */
public class NewsSyncService extends BaseService implements Runnable {
	
	@Resource
	private NewsService newsService;
	
	@Resource
	private ElasticsearchTemplate elasticsearchTemplate;
	
    private static final Logger log = LoggerFactory.getLogger(NewsSyncService.class);
    /**分页参数**/
    private static final int PAGE_SIZE = 1000;
    private int PAGE_NUMBER = 1;
    private int COUNT = 0;
    /**总页数**/
    private int PAGE_TOTAL = 0;
    
    /**是否同步完成**/
    private boolean finished = false ;
    
    private List<News> syncTasks = new ArrayList<News>();

    synchronized public void start() {
            Thread thread = new Thread(this);
            thread.setDaemon(true);
            thread.start();
    }
    
    @Override
    public void run() {
    	while(true){
    	   //是否完成
    	   if(finished) break;
            try {
            	//总数
            	if(COUNT == 0){
            		COUNT = newsService.getNewsCount();
            		
            		PAGE_TOTAL = COUNT % PAGE_SIZE == 0 ? COUNT / PAGE_SIZE : COUNT /PAGE_SIZE + 1;
            		if (PAGE_TOTAL < 1) {
            			PAGE_TOTAL = 1;
            		}
            	}
            	
            	//执行同步任务
                if (PAGE_NUMBER <= PAGE_TOTAL) {
                	 //如果缓冲区为空，则从数据库刷新记录    
                	 HashMap<String,Object> paramMap = new HashMap<String,Object>();
                	 paramMap.put("pageNo", (PAGE_NUMBER -1) *PAGE_SIZE);
                	 paramMap.put("pageSize", PAGE_SIZE);
                	 syncTasks= newsService.getNewsList(paramMap);
                     log.info(String.format("共从数据库中读取了%d条(共%d条)同步任务,总共%d页,当前第%d页", syncTasks.size(), COUNT,PAGE_TOTAL,PAGE_NUMBER));
                     PAGE_NUMBER ++;
                     doSync(syncTasks);
                     Thread.sleep(3000);
                    
                   // Thread.sleep(syncTasks.size() > 0 ? 1000 : 5000); // 防止过度占用CPU
                }else{
                	finished = true;
                } 
            } catch (Throwable e) {
                log.error("", e);
                try {
                    Thread.sleep(15000); // 防止过度占用CPU
                } catch (InterruptedException e1) {
                    log.error("", e1);
                }
            }
    		
    	}
    }
    

    /**
     * 
     * @Title: doSync
     * @Description: 同步创建索引
     * @param syncTasks
     * void
     * @throws Exception boolean
     * @author  kangjie1209@163.com
     * @date 2015年10月20日 下午12:16:23
     */
    private void doSync(List<News> syncTasks) throws Exception {
        /*log.info(String.format("同步脏记录%s(%s)，数据类型[%s] ID[%s] 同步类型[%s(%s)]  当前缓冲记录[%d]",
                result.isSuccess() ? "成功" : "失败", result.getMsg(), task.getDataType(), task.getDataId(),
                task.getSyncType(), task.getSubSyncType(), syncTasks.size()));*/
    	List<IndexQuery> indexQueries = new ArrayList<IndexQuery>();
    	for(News news:syncTasks){
    		IndexQuery indexQuery = new IndexQueryBuilder().withId(String.valueOf(news.getId())).withObject(news).build();
    		indexQueries.add(indexQuery);
    	}
    	elasticsearchTemplate.bulkIndex(indexQueries);
    }

   
}
